package concurrent;

import java.util.Collection;
import java.util.Iterator;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.ReentrantLock;

/**
 * LinkedList 实现的，FIFO 的并发阻塞队列
 *
 * 控制并发，读写各加一把锁。
 * 并根据两把锁，生成两个条件谓词：notEmpty 和 notFull。用于通知take和put线程执行。
 *
 * 顺便说一句：
 * add, put, offer 的区别
 * 在链表满了的情况下有不同表现：
 * add会抛出异常，put会阻塞直到有空间可以放，offer会返回false
 *
 * @Author Antony
 * @Since 2018/6/22 12:10
 */
public class MyLinkedBlockingQueue<E> implements BlockingQueue<E> {

    // 此处注意，头节点只是作为一个起始位置，在初始化的时候就设置了一个item=null的起始节点。
    // 所以 存元素的话，是以这个首节点开始追加的。
    // 所以取元素的话，需要弹出首元素，value在接下来的元素item中。
    private Node<E> head;
    private Node<E> last;

    /** 链表容量, 如果为null的话，则容量为Integer.MAX_VALUE */
    private final int capacity;
    /** 链表元素数 */
    private final AtomicInteger count = new AtomicInteger();

    /** 无参构造器，默认capacity=Integer.MAX_VALUE */
    public MyLinkedBlockingQueue() {
        this(Integer.MAX_VALUE);
    }

    /** 初始化容器 */
    public MyLinkedBlockingQueue(int capacity) {
        if(capacity < 0) throw new IllegalArgumentException();
        this.capacity = capacity;
        last = head = new Node(null);
    }

    private final ReentrantLock putLock = new ReentrantLock(true);
    private final ReentrantLock takeLock = new ReentrantLock();
    private final Condition notEmpty = takeLock.newCondition();
    private final Condition notFull = putLock.newCondition();

    /**
     * Linked linkedlist node class
     */
    static class Node<E>{
        E item;

        /**
         * 表示下一个节点引用
         */
        Node<E> next;
        Node(E e){ item = e;}
    }

    @Override
    public void put(E e) throws InterruptedException {
//        final ReentrantLock putLock = this.putLock;
//        final AtomicInteger count = this.count;
        putLock.lockInterruptibly();
        int c = -1;
        try {
            while(count.get() == capacity){
//                System.out.println("队列满了,put阻塞中...");
                notFull.await();
            }
            enqueue(e);

            c = count.getAndIncrement();
            if(c+1 < capacity ){
                notFull.signal();
            }
        } finally {
            putLock.unlock();
        }

        if(c > 0){
            singalNotEmpty();
        }
    }


    @Override
    public E take() throws InterruptedException {
//        final ReentrantLock takeLock = this.takeLock;
//        final AtomicInteger count = this.count;
        takeLock.lockInterruptibly();
        E item;
        int c = -1;
        try {
            while(count.get() == 0){
//                System.out.println("队列空了,take阻塞中...");
                notEmpty.await();
            }
            item = dequeue();

            c = count.getAndDecrement();
            if(c > 1){
                notEmpty.signal();
            }
        } finally {
            takeLock.unlock();
        }

        if(c == capacity){
            singalNotFull();
        }

        return item;
    }

    /** 向尾节点追加元素 */
    private void enqueue(E e){
        Node<E> node = new Node(e);
        last.next = node;
        last = node;
    }

    /** 从头节点弹出元素,并且把头节点指针引用去掉，帮助GC */
    private E dequeue(){
        Node<E> h = head;
        Node<E> first = head.next;
        h.next = h;
        head = first;
        E item = first.item;
        first.item = null;
        return item;
    }

    /**
     * 获取put锁，然后通知notFull
     */
    private void singalNotFull(){
        final ReentrantLock putLock = this.putLock;
        putLock.lock();
        try {
            notFull.signal();
        } finally {
            putLock.unlock();
        }
    }

    /**
     * 获取take锁，然后通知notEmpty
     */
    private void singalNotEmpty(){
        final ReentrantLock takeLock = this.takeLock;
        takeLock.lock();
        try {
            notEmpty.signal();
        } finally {
            takeLock.unlock();
        }
    }


    @Override
    public boolean add(E e) {
        return false;
    }

    @Override
    public boolean offer(E e) {
        return false;
    }

    @Override
    public E remove() {
        return null;
    }

    @Override
    public E poll() {
        return null;
    }

    @Override
    public E element() {
        return null;
    }

    @Override
    public E peek() {
        return null;
    }

    @Override
    public boolean offer(E e, long timeout, TimeUnit unit) throws InterruptedException {
        return false;
    }

    @Override
    public E poll(long timeout, TimeUnit unit) throws InterruptedException {
        return null;
    }

    @Override
    public int remainingCapacity() {
        return 0;
    }

    @Override
    public boolean remove(Object o) {
        return false;
    }

    @Override
    public boolean containsAll(Collection<?> c) {
        return false;
    }

    @Override
    public boolean addAll(Collection<? extends E> c) {
        return false;
    }

    @Override
    public boolean removeAll(Collection<?> c) {
        return false;
    }

    @Override
    public boolean retainAll(Collection<?> c) {
        return false;
    }

    @Override
    public void clear() {

    }

    @Override
    public int size() {
        return 0;
    }

    @Override
    public boolean isEmpty() {
        return false;
    }

    @Override
    public boolean contains(Object o) {
        return false;
    }

    @Override
    public Iterator<E> iterator() {
        return null;
    }

    @Override
    public Object[] toArray() {
        return new Object[0];
    }

    @Override
    public <T> T[] toArray(T[] a) {
        return null;
    }

    @Override
    public int drainTo(Collection<? super E> c) {
        return 0;
    }

    @Override
    public int drainTo(Collection<? super E> c, int maxElements) {
        return 0;
    }
}
